package com.zzy.broker.handler;

import com.zzy.broker.server.message.MqttRawMessage;
import com.zzy.broker.session.ClientSession;
import com.zzy.broker.session.SessionManager;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.StandardCharsets;
import java.util.List;

public class PublishHandler extends SimpleChannelInboundHandler<MqttRawMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MqttRawMessage msg) throws Exception {
        // 只处理 PUBLISH 消息 (固定报头高4位 = 3)
        byte messageType = (byte) ((msg.header >> 4) & 0x0F);
        
        if (messageType != 3) {  // 3 = PUBLISH
            // 不是 PUBLISH 消息，传递给下一个 Handler
            ctx.fireChannelRead(msg);
            return;
        }
        
        System.out.println("=== MQTT PUBLISH Message ===");
        System.out.println("Fixed Header: 0x" + String.format("%02X", msg.header));
        
        // 解析 PUBLISH 标志位
        boolean dup = ((msg.header & 0x08) != 0);      // 重发标志
        int qos = (msg.header & 0x06) >> 1;            // QoS 等级
        boolean retain = ((msg.header & 0x01) != 0);   // 保留标志
        
        System.out.println("DUP: " + dup + ", QoS: " + qos + ", Retain: " + retain);
        
        ByteBuf payload = Unpooled.wrappedBuffer(msg.payload);
        
        try {
            // 1. 读取主题名称
            if (payload.readableBytes() < 2) {
                throw new IllegalArgumentException("Insufficient data for topic length");
            }
            int topicLen = payload.readUnsignedShort();
            if (payload.readableBytes() < topicLen) {
                throw new IllegalArgumentException("Insufficient data for topic");
            }
            
            byte[] topicBytes = new byte[topicLen];
            payload.readBytes(topicBytes);
            String topic = new String(topicBytes, StandardCharsets.UTF_8);
            System.out.println("Topic: " + topic);
            
            // 2. 如果 QoS > 0，读取 Packet Identifier
            int packetId = 0;
            if (qos > 0) {
                if (payload.readableBytes() < 2) {
                    throw new IllegalArgumentException("Insufficient data for packet identifier");
                }
                packetId = payload.readUnsignedShort();
                System.out.println("Packet ID: " + packetId);
            }
            
            // 3. MQTT 5.0: 跳过 Properties（如果有）
            // 简化处理：这里假设没有 properties 或已经在前面处理
            
            // 4. 读取消息内容（剩余的所有字节）
            int messageLen = payload.readableBytes();
            byte[] messageBytes = new byte[messageLen];
            payload.readBytes(messageBytes);
            String message = new String(messageBytes, StandardCharsets.UTF_8);
            
            System.out.println("Message Length: " + messageLen);
            System.out.println("Message: " + message);
            
            // 5. 根据 QoS 返回确认
            if (qos == 1) {
                // QoS 1: 返回 PUBACK
                byte[] pubackBytes = new byte[]{
                    (byte)0x40,              // 固定报头: PUBACK
                    0x02,                    // 剩余长度: 2字节
                    (byte)(packetId >> 8),   // Packet ID 高字节
                    (byte)(packetId & 0xFF)  // Packet ID 低字节
                };
                System.out.println("Sending PUBACK for Packet ID: " + packetId);
                ctx.writeAndFlush(Unpooled.wrappedBuffer(pubackBytes));
                
            } else if (qos == 2) {
                // QoS 2: 返回 PUBREC（第一步确认）
                byte[] pubrecBytes = new byte[]{
                    (byte)0x50,              // 固定报头: PUBREC
                    0x02,                    // 剩余长度: 2字节
                    (byte)(packetId >> 8),   // Packet ID 高字节
                    (byte)(packetId & 0xFF)  // Packet ID 低字节
                };
                System.out.println("Sending PUBREC for Packet ID: " + packetId);
                ctx.writeAndFlush(Unpooled.wrappedBuffer(pubrecBytes));
            }
            // QoS 0: 不需要确认
            
            // 消息路由 - 将消息转发给订阅了该主题的客户端
            routeMessageToSubscribers(topic, messageBytes, qos, retain);
            
            System.out.println("Message received and routed successfully!");
            
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            payload.release();
        }
    }
    
    /**
     * 将消息路由转发给订阅了相关主题的客户端
     */
    private void routeMessageToSubscribers(String topic, byte[] messagePayload, int publishQos, boolean retain) {
        // 获取订阅了该主题的所有客户端
        List<ClientSession> subscribers = SessionManager.getInstance().getSubscribedSessions(topic);
        
        System.out.println("Routing message to " + subscribers.size() + " subscriber(s) for topic: " + topic);
        
        for (ClientSession subscriber : subscribers) {
            if (!subscriber.isActive()) {
                System.out.println("Skipping inactive subscriber: " + subscriber.getClientId());
                continue;
            }
            
            // 获取订阅者的 QoS，实际发送使用较小的 QoS
            Integer subscriberQos = subscriber.getSubscriptionQos(topic);
            if (subscriberQos == null) {
                // 如果没有直接匹配，需要遍历所有订阅查找匹配的主题
                for (String subscribedTopic : subscriber.getSubscribedTopics()) {
                    if (topicMatches(subscribedTopic, topic)) {
                        subscriberQos = subscriber.getSubscriptionQos(subscribedTopic);
                        break;
                    }
                }
            }
            
            if (subscriberQos == null) {
                continue; // 跳过，不应该发生
            }
            
            int effectiveQos = Math.min(publishQos, subscriberQos);
            
            try {
                // 构造 PUBLISH 消息
                byte[] publishMessage = buildPublishMessage(topic, messagePayload, effectiveQos, retain);
                
                // 发送给订阅者
                subscriber.getChannel().writeAndFlush(Unpooled.wrappedBuffer(publishMessage)).addListener(future -> {
                    if (future.isSuccess()) {
                        System.out.println("Message forwarded to client: " + subscriber.getClientId());
                    } else {
                        System.err.println("Failed to forward message to client: " + subscriber.getClientId() + ", error: " + future.cause());
                    }
                });
                
            } catch (Exception e) {
                System.err.println("Error routing message to " + subscriber.getClientId() + ": " + e.getMessage());
                e.printStackTrace();
            }
        }
    }
    
    /**
     * 构造 PUBLISH 消息
     */
    private byte[] buildPublishMessage(String topic, byte[] payload, int qos, boolean retain) {
        byte[] topicBytes = topic.getBytes(StandardCharsets.UTF_8);
        
        // 计算总长度
        int remainingLength = 2 + topicBytes.length + payload.length; // topic length(2) + topic + payload
        if (qos > 0) {
            remainingLength += 2; // Packet ID
        }
        
        // 构造固定报头
        byte fixedHeader = (byte)0x30; // PUBLISH
        if (retain) {
            fixedHeader |= 0x01; // Retain flag
        }
        fixedHeader |= (qos << 1); // QoS
        
        // 构造完整消息
        ByteBuf buffer = Unpooled.buffer();
        buffer.writeByte(fixedHeader);
        
        // 写入剩余长度（可变长度编码）
        encodeRemainingLength(buffer, remainingLength);
        
        // 写入主题
        buffer.writeShort(topicBytes.length);
        buffer.writeBytes(topicBytes);
        
        // 写入 Packet ID（如果 QoS > 0）
        if (qos > 0) {
            // 生成一个简单的 Packet ID（实际应用中应该维护一个自增 ID）
            int packetId = (int)(System.currentTimeMillis() % 65535);
            buffer.writeShort(packetId);
        }
        
        // 写入消息内容
        buffer.writeBytes(payload);
        
        byte[] result = new byte[buffer.readableBytes()];
        buffer.readBytes(result);
        buffer.release();
        
        return result;
    }
    
    /**
     * 编码剩余长度（MQTT 可变长度编码）
     */
    private void encodeRemainingLength(ByteBuf buffer, int length) {
        do {
            int digit = length % 128;
            length = length / 128;
            if (length > 0) {
                digit = digit | 0x80;
            }
            buffer.writeByte(digit);
        } while (length > 0);
    }
    
    /**
     * 主题匹配（支持 MQTT 通配符）
     */
    private boolean topicMatches(String filter, String topic) {
        if (filter.equals(topic)) {
            return true;
        }
        
        String[] filterLevels = filter.split("/");
        String[] topicLevels = topic.split("/");
        
        int filterIndex = 0;
        int topicIndex = 0;
        
        while (filterIndex < filterLevels.length && topicIndex < topicLevels.length) {
            String filterLevel = filterLevels[filterIndex];
            String topicLevel = topicLevels[topicIndex];
            
            if (filterLevel.equals("#")) {
                return true;
            } else if (filterLevel.equals("+")) {
                filterIndex++;
                topicIndex++;
            } else if (filterLevel.equals(topicLevel)) {
                filterIndex++;
                topicIndex++;
            } else {
                return false;
            }
        }
        
        return filterIndex == filterLevels.length && topicIndex == topicLevels.length;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}
